package Zookeeper.case2;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributedLock {

    private final ZooKeeper zk;
    private final String connectString = "192.168.130.133:2181";
    private final int sessionTimeout = 2000;
    private String waitPath;
    private CountDownLatch connectLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);
    private String currentNode;

    public DistributedLock() throws IOException, InterruptedException, KeeperException {

        // 获取连接
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {

                // connectLatch  如果连接到zk 可以释放
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                    connectLatch.countDown();
                }

                // 发生了 waitPath 的删除事件
                if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
                    waitLatch.countDown();
                }
            }
        });

        // 等待 zk正常连接后 往下走程序
        connectLatch.await();

        // 判断根节点/locks是否存在
        Stat stat = zk.exists("/locks", false);
        if (stat == null) {
            // 表示根节点不存在
            System.out.println("根节点不存在");
            zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    // 对zk加锁
    public void zkLock() {

        // 创建对于的临时带序号节点
        try {
            currentNode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            // wait 一小会, 让结果更清晰一些
            Thread.sleep(10);

            // 判断创建的节点是否时最小的序号节点，如果是则获得锁 否则监听前一个节点
            List<String> children = zk.getChildren("/locks", false);

            // 如果children只有一个值 直接获得锁，否则需要判断谁最小
            if (children.size() == 1) {
                return;
            } else {

                // 对根节点下的所有临时节点排序
                Collections.sort(children);

                // 获取节点名称 seq-0000000000
                String thisNode = currentNode.substring("/locks/".length());

                // 通过seq-0000000获取节点在children集合的位置
                int index = children.indexOf(thisNode);
                if (index == -1) {
                    System.out.println("数据异常！");
                } else if (index == 0) {
                    // 只有一个节点
                    return;
                } else {
                    // 监听他前一个节点变化
                    waitPath = "/locks/" + children.get(index - 1);

                    // 在 waitPath 上注册监听器, 当 waitPath 被删除时, zookeeper 会回调监听器的 process 方法
                    zk.getData(waitPath, true, null);

                    // 等待监听  进入等待锁状态
                    waitLatch.await();
                    return;
                }
            }
        } catch (KeeperException | InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    // 对zk解锁
    public void zkUnlock() throws InterruptedException, KeeperException {

        // 删除节点
        zk.delete(this.currentNode, -1);
    }
}
